/**
 * Dans.com Inc. Copyright (c) 2011-2020 All Rights Reserved
 */

package com.example.mqreceiver;

import com.example.mqreceiver.config.RabbitMqConfig;
import com.rabbitmq.client.Channel;
import event.BusinessEvent;
import notify.Message;
import notify.MessageReceiver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

/**
 * 消息接收处理
 * @author ke
 * Created by on  2021-05-27 10:02
 */
@Service
public class MqEventReceiver implements MessageReceiver {

    @Autowired
    private RabbitMqConfig config;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MqEventProcessorFactory mqEventProcessorFactory;

    private static final Logger log = LoggerFactory.getLogger(MqEventReceiver.class);

    /**
    * 接收消息
    *
    * @param message
    * @return void
    * @author kehong
    * @date 2021-05-26
    */
    @Override
    @Transactional(rollbackFor = Throwable.class)
    public void receiver(Message message) {
        BusinessEvent oldEvent = (BusinessEvent) message;
        BusinessEvent newEvent= mqEventProcessorFactory.getEvent(oldEvent.getEventType().getType());
        if (null == newEvent) {
            return;
        }else{
            newEvent.setBusinessId(oldEvent.getBusinessId());
            newEvent.setEventType(oldEvent.getEventType());
            newEvent.setOperateType(oldEvent.getOperateType());
            newEvent.setBusinessId(oldEvent.getBusinessId());
            newEvent.setData(oldEvent.getData());
        }
        MqEventProcessor processor = mqEventProcessorFactory
                .getEventProcessor(newEvent.getEventType().getType());

        if (null == processor) {
            return;
        }
        processor.action(newEvent);
    }

    /**
     * 特别注意： 如果是通过消息处理事件，必须要避免在消息中查询触发当前消息的事务提交的数据库内容 避免事务提交在消息到达之后生效，从而导致脏读
     *
     * @param bizMessage
     * @param channel
     * @param message
     * @throws Exception
     */

    @RabbitListener(queues = "${rabbitdemo.event.queue.name}")
    public void receiver(Message bizMessage, Channel channel, org.springframework.amqp.core.Message message)
            throws Exception {
        try {
            log.info("receiver message:{}", bizMessage);
            // 启用QoS，每次预取1条消息，避免消息处理不过来，全部堆积在本地缓存里
            channel.basicQos(5);
            receiver(bizMessage);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }  catch (Exception e) {
            log.error("process message failed", bizMessage, e);
            processException(bizMessage, channel, message, e);
        }
    }

    private void processException(Message bizMessage, Channel channel, org.springframework.amqp.core.Message message,
                                  Throwable t) throws Exception {
        Integer retryTime = (Integer) message.getMessageProperties().getHeaders().get(RabbitMqConfig.RABBIT_HEADER_RETRY_KEY);

        if (null != retryTime && retryTime >= config.getRetryTimes()) {
            log.error("bizMsg:{}, msg:{} retry failed, will be deliver to dead queue, should be attention",
                    bizMessage, message.getMessageProperties().getDeliveryTag());
            // 发送到失败的消息队列中
            rabbitTemplate.convertAndSend(config.getFailedExchangeName(), config.getRoutingKey());
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            if (retryTime == null) {
                retryTime = 0;
            }
            retryTime++;
            log.info("retry in {} time, bizMsg:{},msg:{} ", retryTime, bizMessage,
                    message.getMessageProperties().getDeliveryTag());
            // 没有到达重试次数，塞回重试队列
            Integer finalRetryTime = retryTime;
            rabbitTemplate.convertAndSend(config.getRetryExchangeName(), config.getRoutingKey(), bizMessage,
                    message1 -> {
                        message1.getMessageProperties().getHeaders().put(RabbitMqConfig.RABBIT_HEADER_RETRY_KEY, finalRetryTime);
                        return message1;
                    });
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

}

